package io.github.eric.urpc.server.handler;

import io.github.eric.urpc.core.conn.Connection;
import io.github.eric.urpc.core.conn.ConnectionHelper;
import io.github.eric.urpc.core.rpc.*;
import io.github.eric.urpc.core.rpc.msg.VoidResponse;
import io.github.eric.urpc.core.thread.OrderedThreadKeyContext;
import io.github.eric.urpc.core.util.ConstantsUtil;
import io.github.eric.urpc.server.ServerContext;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;

import java.lang.reflect.Method;
import java.util.concurrent.RejectedExecutionException;

@Sharable
public class ServerMsgHandler extends ServerBaseHandler {

	public ServerMsgHandler(ServerBootstrap serverBootstrap, ServerContext serverContext) {
		super(serverBootstrap, serverContext);
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) {
		Message tmpMsg = null;
		try {
			tmpMsg = (Message) msg;
			if (tmpMsg == null) {
				if (serverContext.getLogger().isWarnEnabled()) {
					serverContext.getLogger().warn("[urpc server] receive msg from {}, but class type do not extends '{}', drop it",
							Message.class, ctx.channel().remoteAddress());
				}
				return;
			}
			Message message = tmpMsg;
			OrderedThreadKeyContext.setCurrent(message.getOrderedThreadKey());
			serverContext.getRpcExecutor().execute(new Runnable() {
				@Override
				public void run() {
					try {
						if (Response.class.isAssignableFrom(message.getClass())) {
							// no need to check response registered or not
							ConnectionHelper.processResponse(ctx.channel().attr(Connection.TCP).get(), (Response) message);
						} else {
							// check if register this msg
							MessageRegistrar messageRegistrar = serverContext.getMessageRegistrar();
							MessageInfo messageInfo = messageRegistrar.getMessageInfo(message.getClass());
							Method method = messageInfo.getMethod();
							if (method == null) {
								if (serverContext.getLogger().isWarnEnabled()) {
									serverContext.getLogger().warn("[urpc server] receive msg {} from {}, but can not find rpc method, drop it",
											message.getClass().getName(), ctx.channel().remoteAddress());
								}
								return;
							}
							method.setAccessible(true);
							Object result;
							if (messageInfo.getRpcContextIndex() < 0) {
								result = method.invoke(messageInfo.getService(), message);
							} else if (messageInfo.getRpcContextIndex() == 0) {
								result = method.invoke(messageInfo.getService(), buildRpcContext(ctx), message);
							} else {
								result = method.invoke(messageInfo.getService(), message, buildRpcContext(ctx));
							}
							if (result != null) {
								if (Response.class.isAssignableFrom(result.getClass())) {
									Response response = ((Response) result);
									Request request = ((Request) message);
									response.setRequestId(request.getRequestId());
									response.setOrderedThreadKey(request.getOrderedThreadKey());
								}
								ctx.channel().writeAndFlush(result);
							} else if (Request.class.isAssignableFrom(message.getClass())) {
								Request request = ((Request) message);
								VoidResponse voidResponse = new VoidResponse();
								voidResponse.setRequestId(((Request) message).getRequestId());
								voidResponse.setOrderedThreadKey(request.getOrderedThreadKey());
								ctx.channel().writeAndFlush(voidResponse);
							}
						}
					} catch (Exception e) {
						if (serverContext.getLogger().isWarnEnabled()) {
							serverContext.getLogger().warn("[urpc server] handle msg error, remote addr {}", ctx.channel().remoteAddress(), e);
						}
					}
				}

				@Override
				public String toString() {
					return "serverRecvRpcTask";
				}
			});
		} catch (RejectedExecutionException e) {
			serverContext.getLogger().error("[urpc server] handle msg task is rejected, you can increase rpcExecutorMaxSize or check slow IO, remote addr {}",
					ctx.channel().remoteAddress(), e);
		} finally {
			if (tmpMsg != null) {
				OrderedThreadKeyContext.removeCurrent();
			}
		}
		ctx.fireChannelRead(msg);
	}

	private RpcContext buildRpcContext(ChannelHandlerContext ctx) {
		Connection conn = ctx.channel().attr(Connection.TCP).get();
		return new RpcContext(conn.getId(),
				conn.getAttribute(ConstantsUtil.APPLICATION_NAME) == null ? null : (String) conn.getAttribute(ConstantsUtil.APPLICATION_NAME),
				conn.getAttribute(ConstantsUtil.APPLICATION_ID) == null ? null : (String) conn.getAttribute(ConstantsUtil.APPLICATION_ID),
				ctx.channel().localAddress(), ctx.channel().remoteAddress());
	}
}
